package com.maowei.config;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.google.common.collect.Lists;
import com.maowei.event.TableEvent;
import org.springframework.beans.BeansException;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * canal 客户端连接配置
 *
 * @author maowei
 * @date 2021/8/24
 */
@Configuration
@ConditionalOnClass(CanalConnectors.class)
public class CanalConnectorConfiguration implements ApplicationContextAware {

    private final CanalConfigProperties canalConfigProperties;
    protected ApplicationContext applicationContext;

    public CanalConnectorConfiguration(CanalConfigProperties canalConfigProperties) {
        this.canalConfigProperties = canalConfigProperties;
    }

    @Bean
    public CanalConnector initConnector() {
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalConfigProperties.getHost(), canalConfigProperties.getPort()),
                canalConfigProperties.getDestination(), canalConfigProperties.getUserName(), canalConfigProperties.getPassword());
        connector.connect();
        connector.subscribe(String.join(",", mergeFilters(getConfigFilters(), getTableEventFilters())));
        connector.rollback();
        return connector;
    }

    /**
     * 合并canal需要过滤的表
     *
     * @param configFilters 配置文件
     * @param annotationFilters TableEvent注解
     * @return
     */
    private List<String> mergeFilters(List<String> configFilters, List<String> annotationFilters) {
        List<String> result = new ArrayList<>();
        if (!CollectionUtils.isEmpty(configFilters)) {
            result.addAll(configFilters);
        }
        if (!CollectionUtils.isEmpty(annotationFilters)) {
            result.addAll(annotationFilters);
        }
        Assert.notEmpty(result, "目标过滤表不能为空");
        return result.stream()
                .distinct()
                .collect(Collectors.toList());
    }

    /**
     * 通过配置文件获得canal需要过滤的表
     *
     * @return 过滤器
     */
    private List<String> getConfigFilters() {
        if (StringUtils.isEmpty(canalConfigProperties.getFilter())) {
            return Lists.newLinkedList();
        }
        return Lists.newArrayList(canalConfigProperties.getFilter().split(","));
    }

    /**
     * 通过TableEvent注解获得canal需要过滤的表
     *
     * @return 过滤器
     */
    private List<String> getTableEventFilters() {
        List<TableEvent> tableEventList = new ArrayList<>();
        final Map<String, Object> beansWithAnnotation = applicationContext.getBeansWithAnnotation(TableEvent.class);
        for (Object bean : beansWithAnnotation.values()) {
            TableEvent tableEvent = bean.getClass().getAnnotation(TableEvent.class);
            if (tableEvent != null) {
                tableEventList.add(tableEvent);
            }
        }
        if (CollectionUtils.isEmpty(tableEventList)) {
            return null;
        }
        return tableEventList.stream()
                .map(it -> it.schemaName() + '.' + it.tableName())
                .distinct()
                .collect(Collectors.toList());
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
